package net.sh.dpinfrastructure.message;

import java.util.List;
import java.util.concurrent.Executors;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;

import net.sh.dpinfrastructure.log.DPLogger;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

/**
 * DPMessage基于redis的实现<br/>适用于简单场景，在休息队列模式下不能保证消息被成功消费<br/>
 * 优点是效率较高
 * 
 * @author Lancer
 * @date 2016年4月17日上午11:44:25
 *
 */
public class DPRedisMessage implements DPMessage {
    
    private JedisPool jedisPool;
    
    private final DPLogger logger = new DPLogger(DPRedisMessage.class);
    
    public DPRedisMessage(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    @Override
    public String consume(String queueName) {
        Jedis jedis = jedisPool.getResource();
        try {
            String message = jedis.rpop(queueName);
            return message;
        } finally {
            jedis.close();
        }
    }

    @Override
    public void consumeAsync(String queueName, final Messagelistener messagelistener, final boolean quit4Exception) {
        Jedis jedis = jedisPool.getResource();
        Executors.newSingleThreadExecutor().execute(new Runnable() {
            @Override
            public void run() {
                if (quit4Exception) {
                    try {
                        while (true) {
                            List<String> message = jedis.brpop(ArrayUtils.toArray(queueName, "0"));
                            if (CollectionUtils.isEmpty(message)) {
                                continue;
                            }
                            messagelistener.onMessage(message.get(1));
                        }
                    } catch (Exception e) {
                        logger.error("消费消息异常，已停止消费", e);
                        throw e;
                    } finally {
                        jedis.close();
                    }
                } else {
                    while (true) {
                        try {
                            List<String> message = jedis.brpop(ArrayUtils.toArray(queueName, "0"));
                            if (CollectionUtils.isEmpty(message)) {
                                continue;
                            }
                            messagelistener.onMessage(message.get(1));
                        } catch (Exception e) {
                            logger.error("消费消息异常", e);
                        }
                    }
                }
            }
        });
    }
    
    @Override
    public void consumeSustainBlock(String queueName, Messagelistener messageListener, boolean quit4Exception) {
        Jedis jedis = jedisPool.getResource();
        if (quit4Exception) {
            try {
                while (true) {
                    List<String> message = jedis.brpop(ArrayUtils.toArray(queueName, "0"));
                    if (CollectionUtils.isEmpty(message)) {
                        continue;
                    }
                    messageListener.onMessage(message.get(1));
                }
            } catch (Exception e) {
                throw e;
            } finally {
                jedis.close();
            }
        } else {
            while (true) {
                try {
                    List<String> message = jedis.brpop(ArrayUtils.toArray(queueName, "0"));
                    if (CollectionUtils.isEmpty(message)) {
                        continue;
                    }
                    messageListener.onMessage(message.get(1));
                } catch (Exception e) {
                    logger.error("消费消息异常", e);
                }
            }
        }
    }
    
    @Override
    public void produce(String queueName, String message) {
        Jedis jedis = jedisPool.getResource();
        try {
            jedis.lpush(queueName, message);
        } finally {
            jedis.close();
        }
    }

    @Override
    public void publish(String channel, String message) {
        Jedis jedis = jedisPool.getResource();
        try {
            jedis.publish(channel, message);
        } finally {
            jedis.close();
        }
    }

    @Override
    public void subscribe(String channel, String message, Messagelistener msgListener, boolean unsubscribeOnException) {
        Jedis jedis = jedisPool.getResource();
        try {
            jedis.subscribe(new JedisPubSub() {
                
                @Override
                public void onMessage(String channel, String message) {
                    try {
                        msgListener.onMessage(message);
                    } catch (Throwable t) {
                        if (unsubscribeOnException) {
                            throw t;
                        }
                        logger.error("订阅消息失败：" + message, t);
                    }
                }
                
            }, channel);
        } finally {
            jedis.close();
        }
    }
}
